/*
 * Created by Gibbs on 2021/1/1.
 * Copyright (c) 2021 Gibbs. All rights reserved.
 */

#include <base/Log.h>
#include <thread>
#include "MessageSource.h"
#include "Source.h"

#define TAG "MessageSource"

MessageSource::MessageSource() {
    LOGI("MessageSource", "CoreFlow : create MessageSource");
    msgQueue = new ConcurrentQueue<Message *>(1000000, 0, "MessageQueue");
}

MessageSource::~MessageSource() {
    LOGI("MessageSource", "CoreFlow : MessageSource destroyed %d", msgQueue->size());
    delete msgQueue;
    msgQueue = nullptr;
}

unsigned long MessageSource::pushMessage(int from, int type, long extra) {
    unsigned long error;
    if (from == MSG_DOMAIN_TIME) {
        if (currentPosition == type) {
            return SOURCE_RICH;
        }
        currentPosition = type;
    } else if (from == MSG_DOMAIN_BUFFER) {
        bool buffering = (from <= 0 || type <= 0);
        if (buffering == isBuffering) {
            return SOURCE_RICH;
        }
        isBuffering = buffering;
    }
    auto message = new Message(from, type, extra);
    error = msgQueue->push(message);
    LOGI(TAG, "pushMessage %s(%d) %d", message->toString(), message->type, msgQueue->size());
    if (isQueueQuit(error)) {
        return SOURCE_ERROR;
    } else {
        return SOURCE_RICH;
    }
}

unsigned long MessageSource::readMessage(Message **message) {
    Message *msg;
    unsigned long size = msgQueue->front(&msg);
    if (isQueueQuit(size)) {
        return SOURCE_ERROR;
    }
    if (size > 0) {
        (*message)->reset(msg);
        msgQueue->pop();
        return SOURCE_RICH;
    }
    LOGD(TAG, "readMessage size = %d", msgQueue->size());
    return SOURCE_EMPTY;
}

void MessageSource::flush() {
    msgQueue->flush();
    LOGI(TAG, "flushBuffer");
}

void MessageSource::quit() {
    msgQueue->quit();
}

unsigned long MessageSource::size() {
    return msgQueue->size();
}
